﻿/*
-----------------------------------------------------------------------------
This source file is part of Cell Cloud.

Copyright (c) 2009-2012 Cell Cloud Team - cellcloudproject@gmail.com

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
-----------------------------------------------------------------------------
*/

#include "CellStableHeaders.h"
#include "CellNonblockingAcceptor.h"
#include "CellInet4Address.h"
#include "CellNonblockingAcceptorThread.h"
#include "CellMessage.h"
#include "CellUtil.h"

namespace cell
{

#define NBA_MAX_CONNECTNUM 100

//-----------------------------------------------------------------------
NonblockingAcceptor::NonblockingAcceptor(void)
	: MessageAcceptor()
	, mMaxConnectNum(NBA_MAX_CONNECTNUM)
	, mCurrentConnectNum(0)
	, mDispatchThread(NULL)
	, mListener(INVALID_SOCKET)
	, mBase(NULL)
	, mListenerEvent(NULL)
{
	mSessionTimeout.tv_sec = 30;
	mSessionTimeout.tv_usec = 0;

	mWorkers.push_back(CELL_NEW NonblockingWorker(this));
	mWorkers.push_back(CELL_NEW NonblockingWorker(this));
	mWorkers.push_back(CELL_NEW NonblockingWorker(this));
	mWorkers.push_back(CELL_NEW NonblockingWorker(this));

	WorkerList::iterator iter, end = mWorkers.end();
	for (iter = mWorkers.begin(); iter != end; ++iter)
	{
		(*iter)->start();
	}
}
//-----------------------------------------------------------------------
NonblockingAcceptor::~NonblockingAcceptor(void)
{
	unbind();

	clearSessions();

	if (NULL != mDispatchThread)
	{
		CELL_DELETE mDispatchThread;
		mDispatchThread = NULL;
	}

	WorkerList::iterator iter, end = mWorkers.end();
	for (iter = mWorkers.begin(); iter != end; ++iter)
	{
		(*iter)->stop();
	}

	Thread::sleep(100);

	end = mWorkers.end();
	for (iter = mWorkers.begin(); iter != end; ++iter)
	{
		CELL_DELETE (*iter);
	}

	mWorkers.clear();
}
//-----------------------------------------------------------------------
bool NonblockingAcceptor::bind(uint port)
{
	Inet4Address address(port);
	return bind(address);
}
//-----------------------------------------------------------------------
bool NonblockingAcceptor::bind(const InetAddress& addr)
{
	if (INVALID_SOCKET != mListener)
	{
		fireErrorOccurred(MessageHandler::EC_STATE_ERROR, NULL);
		return false;
	}

	mListener = ::socket(AF_INET, SOCK_STREAM, 0);
	if (INVALID_SOCKET == mListener)
	{
		fireErrorOccurred(MessageHandler::EC_SOCK_FAILED, NULL);
		return false;
	}

	// 设置为非阻塞
	evutil_make_socket_nonblocking(mListener);

#ifndef WIN32
	{
		int one = 1;
		setsockopt(mAcceptor->mListener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
	}
#endif

	mPort = addr.getPort();

	if (addr.isIPv4())
	{
		sockaddr_in sockAddr;

		memset(&sockAddr, 0, sizeof(sockAddr));
		sockAddr.sin_family = AF_INET;
		sockAddr.sin_port = htons(addr.getPort());
		if (addr.getHostAddress().empty())
			sockAddr.sin_addr.s_addr = htonl(INADDR_ANY);
		else
			sockAddr.sin_addr.s_addr = inet_addr(addr.getHostAddress().c_str());

		// 绑定
		if (::bind(mListener, (struct sockaddr *)&sockAddr, sizeof(sockAddr)) < 0)
		{
			fireErrorOccurred(MessageHandler::EC_BIND_FAILED, NULL);
			evutil_closesocket(mListener);
			mListener = INVALID_SOCKET;
			return false;
		}
	}
	else
	{
		sockaddr_in6 sockAddr6;

		memset(&sockAddr6, 0, sizeof(sockAddr6));
		sockAddr6.sin6_family = AF_INET6;
		sockAddr6.sin6_port = htons(addr.getPort());
		unsigned long destAddr;
		if (addr.getHostAddress().empty())
			destAddr = INADDR_ANY;
		else
			destAddr = inet_addr(addr.getHostAddress().c_str());
		memcpy(&sockAddr6.sin6_addr, &destAddr, sizeof(struct in6_addr));

		// 绑定
		if (::bind(mListener, (struct sockaddr *)&sockAddr6, sizeof(sockAddr6)) < 0)
		{
			fireErrorOccurred(MessageHandler::EC_BIND_FAILED, NULL);
			evutil_closesocket(mListener);
			mListener = INVALID_SOCKET;
			return false;
		}
	}

	if (::listen(mListener, SOMAXCONN) < 0)
	{
		fireErrorOccurred(MessageHandler::EC_LISTEN_FAILED, NULL);
		evutil_closesocket(mListener);
		mListener = INVALID_SOCKET;
		return false;
	}

	mBase = event_base_new();
	if (NULL == mBase)
	{
		evutil_closesocket(mListener);
		mListener = INVALID_SOCKET;
		return false;
	}

	mListenerEvent = event_new(mBase, mListener, EV_READ|EV_PERSIST, listener_cb, (void*)this);
	event_add(mListenerEvent, NULL);

	// 创建事件循环线程
	if (NULL != mDispatchThread)
	{
		CELL_DELETE mDispatchThread;
		mDispatchThread = NULL;
	}

	mDispatchThread = CELL_NEW NonblockingAcceptorThread(this);
	if (false == mDispatchThread->start())
	{
		evutil_closesocket(mListener);
		mListener = INVALID_SOCKET;
		return false;
	}

	return true;
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::unbind(void)
{
	if (mListener == INVALID_SOCKET)
		return;

	if (NULL != mDispatchThread)
	{
		if (mDispatchThread->isRunning())
		{
			mDispatchThread->stop();
		}

		Thread::sleep(100);

		int count = 0;
		while (mDispatchThread->isRunning())
		{
			Thread::sleep(5);

			++count;
			if (count > 1000)
			{
				break;
			}
		}
	}

	{
		CELL_LOCK_MUTEX(mSessionMutex);

		if (false == mSessions.empty())
		{
			uint64 time = Util::currentTimeMillis();
			SessionMap::iterator iter, end = mSessions.end();
			for (iter = mSessions.begin(); iter != end; ++iter)
			{
				evutil_closesocket(iter->first);
				(iter->second)->closed = true;
				(iter->second)->desertedTime = time;
				mDesertedSessions.push_back(iter->second);
			}
			mSessions.clear();

			mCurrentConnectNum = 0;
		}
	}

	{
		CELL_LOCK_MUTEX(mMonitor);

		if (NULL != mListenerEvent)
		{
			event_free(mListenerEvent);
			mListenerEvent = NULL;
		}
	}

	// 关闭 Socket
	evutil_closesocket(mListener);
	mListener = INVALID_SOCKET;
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::close(Session* session)
{
	NonblockingAcceptorSession* nas = dynamic_cast<NonblockingAcceptorSession*>(session);

	// 关闭
	fireSessionClosed(nas);

	{
		CELL_LOCK_MUTEX(mSessionMutex);

		if (mSessions.find(nas->socket) != mSessions.end())
		{
			if (!nas->bevClosed)
			{
				bufferevent_free(nas->bev);
				nas->bevClosed = true;
			}
		}
	}

	// 销毁
	fireSessionDestroyed(nas);
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::listSessionIds(SessionIdList& list)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	SessionMap::const_iterator iter, end = mSessions.end();
	for (iter = mSessions.begin(); iter != end; ++iter)
	{
		list.push_back(iter->second->getId());
	}
}
//-----------------------------------------------------------------------
Session* NonblockingAcceptor::getSession(ulong id)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	SessionMap::const_iterator iter = mSessions.begin(), end = mSessions.end();
	while (iter != end)
	{
		if (iter->second->getId() == id)
		{
			return iter->second;
		}

		++iter;
	}

	return NULL;
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::setMaxConnectNum(int num)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	mMaxConnectNum = num;
}
//-----------------------------------------------------------------------
int NonblockingAcceptor::getMaxConnectNum(void)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	return mMaxConnectNum;
}
//-----------------------------------------------------------------------
inline int NonblockingAcceptor::getConnectNum(void)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	return mCurrentConnectNum;
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::write(Session* session, Message* message)
{
	NonblockingAcceptorSession* nas = dynamic_cast<NonblockingAcceptorSession*>(session);

	{
		CELL_LOCK_MUTEX(mSessionMutex);

		SessionMap::const_iterator iter = mSessions.find(nas->socket);
		if (iter == mSessions.end())
		{
			return;
		}
	}

	char data[NBA_BUFFER_SIZE] = {0x0};
	uint32 dataSize = message->get(data);

	if (dataSize == 0)
	{
		fireErrorOccurred(MessageHandler::EC_WRITE_FAILED, nas);
		return;
	}

	char* buf = NULL;
	int size = 0;

	if (existDataMark())
	{
		size = dataSize + getHeadLength() + getTailLength();
		buf = CELL_ALLOC_T(char, size + 1, MEMCATEGORY_GENERAL);
		memset(buf, 0x0, size + 1);
		memcpy(buf, getHeadMark(), getHeadLength());
		memcpy(buf + getHeadLength(), data, dataSize);
		memcpy(buf + getHeadLength() + dataSize, getTailMark(), getTailLength());
	}
	else
	{
		size = dataSize;
		buf = CELL_ALLOC_T(char, size + 1, MEMCATEGORY_GENERAL);
		memset(buf, 0x0, size + 1);
		memcpy(buf, data, size);
	}

	bool error = false;

	{
		CELL_LOCK_MUTEX(nas->writeMutex);

		socket_t sock = nas->socket;

		int ssize = 0;
		int ss = 0;
		while (ssize < size)
		{
			if (SOCKET_ERROR == (ss = ::send(sock, buf + ssize, size - ssize, 0)))
			{
				error = true;
				break;
			}

			ssize += ss;
		}
	}

	CELL_FREE(buf, MEMCATEGORY_GENERAL);

	if (false == error)
	{
		fireMessageSent(nas, message);
	}
	else
	{
		fireErrorOccurred(MessageHandler::EC_WRITE_FAILED, nas);
	}
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::read(Message* message, Session* session)
{
	// Nothing
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::setTimeout(int timeInSeconds)
{
	mSessionTimeout.tv_sec = timeInSeconds;
	mSessionTimeout.tv_usec = 0;
}
//-----------------------------------------------------------------------
inline uint NonblockingAcceptor::getPort(void) const
{
	return mPort;
}
//-----------------------------------------------------------------------
NonblockingAcceptorSession* NonblockingAcceptor::findSession(evutil_socket_t fd)
{
	socket_t socket = fd;

	CELL_LOCK_MUTEX(mSessionMutex);

	SessionMap::const_iterator iter = mSessions.find(socket);

	if (iter != mSessions.end())
	{
		return iter->second;
	}
	else
	{
		return NULL;
	}
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::present(NonblockingAcceptorSession* session)
{
	NonblockingWorker* woker = NULL;

	{
		CELL_LOCK_MUTEX(mWorkerMutex);

		WorkerList::const_iterator iter, end = mWorkers.end();
		for (iter = mWorkers.begin(); iter != end; ++iter)
		{
			if ((*iter)->has(session))
			{
				woker = (*iter);
				break;
			}
		}
	}

	if (NULL != woker)
	{
		woker->tread();
	}
}
//-----------------------------------------------------------------------
NonblockingAcceptorSession* NonblockingAcceptor::fireSessionCreated(socket_t socket, const char* host, const uint port)
{
	NonblockingAcceptorSession* session = NULL;

	{
		CELL_LOCK_MUTEX(mSessionMutex);

		Inet4Address address(host, port);
		session = CELL_NEW NonblockingAcceptorSession(this, socket, address);
		mSessions.insert(SessionMap::value_type(socket, session));
	}

	{
		CELL_LOCK_MUTEX(mWorkerMutex);

		int min = mMaxConnectNum;
		int count = 0;
		int index = 0;
		WorkerList::const_iterator iter, end = mWorkers.end();
		for (iter = mWorkers.begin(); iter != end; ++iter)
		{
			NonblockingWorker* worker = (*iter);
			if (worker->capacity() < min)
			{
				min = worker->capacity();
				index = count;
			}
			++count;
		}

		mWorkers[index]->management(session);
	}

	if (NULL != mHandler)
	{
		mHandler->sessionCreated(session);
	}

	return session;
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::fireSessionDestroyed(NonblockingAcceptorSession* session)
{
	{
		CELL_LOCK_MUTEX(mWorkerMutex);

		WorkerList::const_iterator iter, end = mWorkers.end();
		for (iter = mWorkers.begin(); iter != end; ++iter)
		{
			if ((*iter)->unmanagement(session))
			{
				break;
			}
		}
	}

	{
		CELL_LOCK_MUTEX(mSessionMutex);

		SessionMap::iterator iter = mSessions.find(session->socket);
		if (iter != mSessions.end())
		{
			mSessions.erase(iter);
		}

		uint64 time = Util::currentTimeMillis();

		NonblockingAcceptor::SessionList list;

		NonblockingAcceptor::SessionList::iterator siter, send = mDesertedSessions.end();
		for (siter = mDesertedSessions.begin(); siter != send; ++siter)
		{
			if (time - (*siter)->desertedTime > 60000)
			{
				list.push_back((*siter));
			}
		}

		if (false == list.empty())
		{
			send = list.end();
			for (siter = list.begin(); siter != send; ++siter)
			{
				NonblockingAcceptorSession* session = (*siter);

				NonblockingAcceptor::SessionList::iterator sfind = std::find(mDesertedSessions.begin(), mDesertedSessions.end(), session);
				if (sfind != mDesertedSessions.end())
					mDesertedSessions.erase(sfind);

				if (!session->closed)
				{
					evutil_closesocket(session->socket);
					session->closed = true;
				}

				CELL_DELETE session;
			}

			list.clear();
		}

		siter = std::find(mDesertedSessions.begin(), mDesertedSessions.end(), session);
		if (siter == mDesertedSessions.end())
		{
			session->desertedTime = time;
			mDesertedSessions.push_back(session);
		}
	}

	if (NULL != mHandler)
	{
		mHandler->sessionDestroyed(session);
	}
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::fireSessionOpened(NonblockingAcceptorSession* session)
{
	++mCurrentConnectNum;

	if (NULL != mHandler)
	{
		mHandler->sessionOpened(session);
	}
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::fireSessionClosed(NonblockingAcceptorSession* session)
{
	{
		CELL_LOCK_MUTEX(mSessionMutex);

		SessionMap::iterator iter = mSessions.find(session->socket);
		if (iter != mSessions.end())
		{
			mSessions.erase(iter);

			if (!session->bevClosed)
			{
				bufferevent_free(session->bev);
				session->bevClosed = true;
			}
		}

		NonblockingAcceptor::SessionList::iterator siter = std::find(mDesertedSessions.begin(), mDesertedSessions.end(), session);
		if (siter == mDesertedSessions.end())
		{
			uint64 time = Util::currentTimeMillis();
			session->desertedTime = time;
			mDesertedSessions.push_back(session);
		}
	}

	--mCurrentConnectNum;

	if (NULL != mHandler)
	{
		mHandler->sessionClosed(session);
	}
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::fireMessageReceived(NonblockingAcceptorSession* session, Message* message)
{
	if (NULL != mHandler)
	{
		mHandler->messageReceived(session, message);
	}
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::fireMessageSent(NonblockingAcceptorSession* session, Message* message)
{
	if (NULL != mHandler)
	{
		mHandler->messageSent(session, message);
	}
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::fireErrorOccurred(uint16 errorCode, NonblockingAcceptorSession* session)
{
	if (NULL != mHandler)
	{
		mHandler->errorOccurred(errorCode, static_cast<Session*>(session));
	}
}
//-----------------------------------------------------------------------
void NonblockingAcceptor::clearSessions(void)
{
	{
		CELL_LOCK_MUTEX(mSessionMutex);

		if (false == mSessions.empty())
		{
			uint64 time = Util::currentTimeMillis();
			SessionMap::iterator iter, end = mSessions.end();
			for (iter = mSessions.begin(); iter != end; ++iter)
			{
				if (false == (iter->second)->closed)
				{
					evutil_closesocket((iter->second)->socket);
					(iter->second)->closed = true;
				}

				(iter->second)->desertedTime = time;
				mDesertedSessions.push_back(iter->second);
			}
			mSessions.clear();
		}
	}

	if (false == mDesertedSessions.empty())
	{
		Thread::sleep(5);

		CELL_LOCK_MUTEX(mSessionMutex);

		NonblockingAcceptor::SessionList::iterator iter, end = mDesertedSessions.end();
		for (iter = mDesertedSessions.begin(); iter != end; ++iter)
		{
			CELL_DELETE (*iter);
		}

		mDesertedSessions.clear();
	}
}



//-----------------------------------------------------------------------
NonblockingAcceptorSession::NonblockingAcceptorSession(NonblockingAcceptor* acceptor,
	socket_t sock,const InetAddress& address)
	: Session(acceptor, address)
	, mAcceptor(acceptor)
	, mCacheCursor(0)
	, socket(sock)
	, bev(NULL)
	, desertedTime(0)
	, closed(false)
	, bevClosed(false)
{
	memset(mCache, 0x0, NBA_BUFFER_SIZE);
}
//-----------------------------------------------------------------------
NonblockingAcceptorSession::~NonblockingAcceptorSession(void)
{
	{
		CELL_LOCK_MUTEX(mQueueMutex);

		while (false == mDeserted.empty())
		{
			std::list<char*>::iterator iter = mDeserted.begin();
			char* desbuf = (*iter);
			mDeserted.pop_front();
			CELL_FREE(desbuf, MEMCATEGORY_GENERAL);
		}

		while (false == mDataQueue.empty())
		{
			std::list<char*>::iterator iter = mDataQueue.begin();
			char* desbuf = (*iter);
			mDataQueue.pop_front();
			CELL_FREE(desbuf, MEMCATEGORY_GENERAL);
		}

		mLengthQueue.clear();
	}

	if (!bevClosed)
	{
		bufferevent_free(bev);
		bevClosed = true;
	}

	if (!closed)
	{
		evutil_closesocket(socket);
		closed = true;
	}
}
//-----------------------------------------------------------------------
void NonblockingAcceptorSession::pushBuf(char* data, int len)
{
	CELL_LOCK_MUTEX(mQueueMutex);

	char* buf = CELL_ALLOC_T(char, len + 1, MEMCATEGORY_GENERAL);
	memset(buf, 0x0, len + 1);
	memcpy(buf, data, len);

	mDataQueue.push_back(buf);
	mLengthQueue.push_back(len);
}
//-----------------------------------------------------------------------
int NonblockingAcceptorSession::popBuf(char* data)
{
	CELL_LOCK_MUTEX(mQueueMutex);

	while (false == mDeserted.empty())
	{
		std::list<char*>::iterator iter = mDeserted.begin();
		char* desbuf = (*iter);
		mDeserted.pop_front();
		CELL_FREE(desbuf, MEMCATEGORY_GENERAL);
	}

	if (mDataQueue.empty())
	{
		return 0;
	}

	std::list<char*>::iterator dataiter = mDataQueue.begin();
	char* buf = (*dataiter);
	mDataQueue.pop_front();

	std::list<int>::iterator leniter = mLengthQueue.begin();
	int len = (*leniter);
	mLengthQueue.pop_front();

	memcpy(data, buf, len);

	mDeserted.push_back(buf);

	return len;
}
//-----------------------------------------------------------------------
void NonblockingAcceptorSession::notifyRead(void)
{
	int len = 0;
	char data[NBA_BUFFER_SIZE] = {0x0};
	while ((len = popBuf(data)) > 0)
	{
		processRecv(data, len);
	}
}
//-----------------------------------------------------------------------
void NonblockingAcceptorSession::processRecv(char* data, int size)
{
	if (mAcceptor->existDataMark())
	{
		if (mCacheCursor > 0 || (size_t)size <= mAcceptor->getHeadLength() + mAcceptor->getTailLength())
		{
			// 缓存

			int extractedLen = 0;

			if (mCacheCursor + size >= NBA_BUFFER_SIZE)
			{
				extractedLen = extractData(mCache, mCacheCursor);
			}
			else
			{
				char* cache = mCache;
				cache += mCacheCursor;
				memcpy(cache, data, size);
				mCacheCursor += size;

				extractedLen = extractData(mCache, mCacheCursor);
			}

			if (extractedLen > 0)
			{
				memset(mCache, 0x0, NBA_BUFFER_SIZE);
				mCacheCursor = 0;
			}
			else
			{
				memset(mCache, 0x0, NBA_BUFFER_SIZE);
				mCacheCursor = 0;
			}
		}
		else
		{
			int extractedLen = extractData(data, size);
			if (extractedLen < size)
			{
				memcpy(mCache + mCacheCursor, data + extractedLen, size - extractedLen);
				mCacheCursor += size - extractedLen;
			}
		}
	}
	else
	{
		Message msg(data, size);
		mAcceptor->fireMessageReceived(this, &msg);
	}
}
//-----------------------------------------------------------------------
int NonblockingAcceptorSession::extractData(char* data, const int size)
{
	int extractedCursor = 0;

	char* dataptr = data;

	bool incomplete = false;

	while (!incomplete)
	{
		if (extractedCursor > size)
			break;

		bool valid = true;

		for (int i = 0, len = (int)mAcceptor->getHeadLength(); i < len; ++i)
		{
			if (mAcceptor->getHeadMark()[i] != dataptr[extractedCursor + i])
			{
				valid = false;
				break;
			}
		}

		if (valid)
		{
			extractedCursor += mAcceptor->getTailLength();

			int buflen = 0;

			// 开始解析数据
			while (true)
			{
				if (extractedCursor + buflen > size)
				{
					extractedCursor -= mAcceptor->getTailLength();
					incomplete = true;
					break;
				}

				// 判断结尾
				bool end = true;
				for (int i = 0, len = (int)mAcceptor->getTailLength(); i < len; ++i)
				{
					if (mAcceptor->getTailMark()[i] != dataptr[extractedCursor + buflen + i])
					{
						end = false;
						break;
					}
				}

				if (false == end)
				{
					// 数据没有结尾
					++buflen;
				}
				else
				{
					char buf[NBA_BUFFER_SIZE] = {0x0};
					memcpy(buf, dataptr + extractedCursor, buflen);

					Message msg(buf, buflen);
					mAcceptor->fireMessageReceived(this, &msg);

					extractedCursor += buflen;
					extractedCursor += mAcceptor->getTailLength();
					break;
				}
			}
		}
		else
		{
			break;
		}
	}

	return extractedCursor;
}



//-----------------------------------------------------------------------
NonblockingWorker::NonblockingWorker(NonblockingAcceptor* acceptor)
	: mAcceptor(acceptor)
{
}
//-----------------------------------------------------------------------
NonblockingWorker::~NonblockingWorker(void)
{
}
//-----------------------------------------------------------------------
void NonblockingWorker::run(void)
{
	mSpinning = true;

	while (mSpinning)
	{
		{
			SessionList list;

			CELL_LOCK_MUTEX_NAMED(mMutex, lock);

			{
				CELL_LOCK_MUTEX(mSessionMutex);

				SessionList::const_iterator iter, end = mSessions.end();
				for (iter = mSessions.begin(); iter != end; ++iter)
				{
					list.push_back((*iter));
				}
			}

			SessionList::const_iterator iter, end = list.end();
			for (iter = list.begin(); iter != end; ++iter)
			{
				(*iter)->notifyRead();
			}

			list.clear();

			CELL_THREAD_WAIT(mWorkSync, mutex, lock);
		}
	}
}
//-----------------------------------------------------------------------
void NonblockingWorker::stop(void)
{
	CELL_LOCK_MUTEX(mMutex);

	mSpinning = false;

	CELL_THREAD_NOTIFY_ONE(mWorkSync);
}
//-----------------------------------------------------------------------
void NonblockingWorker::management(NonblockingAcceptorSession* session)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	mSessions.push_back(session);
}
//-----------------------------------------------------------------------
bool NonblockingWorker::unmanagement(NonblockingAcceptorSession* session)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	SessionList::const_iterator iter = std::find(mSessions.begin(), mSessions.end(), session);
	if (iter != mSessions.end())
	{
		mSessions.erase(iter);
		return true;
	}

	return false;
}
//-----------------------------------------------------------------------
bool NonblockingWorker::has(NonblockingAcceptorSession* session)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	SessionList::const_iterator iter = std::find(mSessions.begin(), mSessions.end(), session);
	if (iter != mSessions.end())
	{
		return true;
	}
	else
	{
		return false;
	}
}
//-----------------------------------------------------------------------
void NonblockingWorker::tread(void)
{
	CELL_LOCK_MUTEX(mMutex);
	CELL_THREAD_NOTIFY_ONE(mWorkSync);
}
//-----------------------------------------------------------------------
int NonblockingWorker::capacity(void)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	return mSessions.size();
}



//-----------------------------------------------------------------------
void listener_cb(evutil_socket_t listener, short event, void *arg)
{
	struct sockaddr_storage ss;
	socklen_t slen = sizeof(ss);

	evutil_socket_t fd = ::accept(listener, (struct sockaddr*)&ss, &slen);
	if (fd < 0 || fd == INVALID_SOCKET)
	{
		return;
	}

	NonblockingAcceptor* acceptor = static_cast<NonblockingAcceptor*>(arg);

	if (acceptor->mCurrentConnectNum >= acceptor->mMaxConnectNum
		|| acceptor->latch)
	{
		evutil_closesocket(fd);
		return;
	}

	evutil_make_socket_nonblocking(fd);

	struct event_base *base = acceptor->mBase;

	NonblockingAcceptorSession* session = NULL;
	
#if CELL_PLATFORM == CELL_PLATFORM_WIN32
	char clienthost[NI_MAXHOST] = {0x0};
	char clientservice[NI_MAXSERV] = {0x0};
	getnameinfo((sockaddr *)&ss, slen, clienthost, sizeof(clienthost), clientservice, sizeof(clientservice), NI_NUMERICHOST|NI_NUMERICSERV);
	session = acceptor->fireSessionCreated(fd, clienthost, atoi(clientservice));
#else
	// TODO
#endif

	if (NULL != session)
	{
		struct bufferevent *bev = bufferevent_socket_new(base, fd, 0);
		bufferevent_set_timeouts(bev, &acceptor->mSessionTimeout, &acceptor->mSessionTimeout);
		bufferevent_setcb(bev, cell::read_cb, cell::write_cb, cell::event_cb, acceptor);
		bufferevent_enable(bev, EV_READ | EV_WRITE | EV_PERSIST);

		session->bev = bev;

		acceptor->fireSessionOpened(session);
	}
	else
	{
		evutil_closesocket(fd);
	}
}
//-----------------------------------------------------------------------
void read_cb(struct bufferevent *bev, void *arg)
{
	NonblockingAcceptor* acceptor = static_cast<NonblockingAcceptor*>(arg);
	evutil_socket_t fd = bufferevent_getfd(bev);

	NonblockingAcceptorSession* session = acceptor->findSession(fd);
	if (NULL == session)
	{
		return;
	}

	char buf[NBA_BUFFER_SIZE] = {0x0};
	int n = 0;
	while ((n = bufferevent_read(bev, buf, NBA_BUFFER_SIZE)) > 0)
	{
		session->pushBuf(buf, n);
		memset(buf, 0x0, sizeof(buf));
	}

	acceptor->present(session);
}
//-----------------------------------------------------------------------
void write_cb(struct bufferevent *bev, void *arg)
{
}
//-----------------------------------------------------------------------
void event_cb(struct bufferevent *bev, short event, void *arg)
{
	NonblockingAcceptor* acceptor = static_cast<NonblockingAcceptor*>(arg);
	evutil_socket_t fd = bufferevent_getfd(bev);

	NonblockingAcceptorSession* session = acceptor->findSession(fd);

	if (NULL != session)
	{
		if (event & BEV_EVENT_EOF)
		{
			acceptor->fireSessionClosed(session);

			acceptor->fireSessionDestroyed(session);
		}
		else if (event & BEV_EVENT_ERROR)
		{
			acceptor->fireSessionClosed(session);

			acceptor->fireSessionDestroyed(session);
		}
		else if (event & BEV_EVENT_TIMEOUT)
		{
			acceptor->fireSessionClosed(session);

			acceptor->fireSessionDestroyed(session);
		}
	}
}

} // end namespace cell
